package com.wondertek.poms.bc.manager;

import java.io.ByteArrayOutputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.wondertek.poms.bc.feign.KafkaFeignClient;
import com.wondertek.poms.bc.vo.OutSourceDataParamsVo;
import com.wondertek.poms.common.constant.Constants;
import com.wondertek.poms.common.constant.FtlConstants;
import com.wondertek.poms.common.exception.BusinessException;
import com.wondertek.poms.common.exception.MessageCodes;
import com.wondertek.poms.core.service.IContentService;
import com.wondertek.poms.core.service.IPrdContService;
import com.wondertek.poms.core.service.IPrdInfoService;
import com.wondertek.poms.core.utils.ExportXmlUtil;
import com.wondertek.poms.core.utils.OutHttpUtil;
import com.wondertek.poms.dao.po.Content;
import com.wondertek.poms.dao.po.PrdInfo;
import com.wondertek.poms.dao.pojo.ProductContentPojo;

import cn.hutool.core.io.IoUtil;
import cn.hutool.core.util.StrUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * <p><b>StartProcess Description:</b> 同步入包主流程</p>
 * @author douzi
 * <b>DATE</b> 2020年11月12日 下午2:51:13
 */
@Slf4j
@Component
@Data
public class StartProcess {
	@Autowired
	OutHttpUtil outHttpUtil;
    @Autowired
    IContentService contentService;
    @Autowired
    IPrdInfoService prdInfoService;
    @Autowired
    IPrdContService prdContService;
    @Autowired
    KafkaFeignClient kafkaFeignClient;
	
//    @Async(value = "threadPoolA")
    public void fastProcess(OutSourceDataParamsVo paramVo) {
        process(paramVo);
    }

    /**
	 * <p><b>Title:</b> process</p>
	 * <p><b>Description:</b> 主业务流程 </p>
	 * type: 0浏览,1点播，2直播，3下载，4点播+下载，5模拟直播，6内容集，7精简编码，8精简直播，9精简模拟直播，10：剧集 12互联网直播
	 * formtype: 6：剧集（壳）7：剧集的单集8：非剧集9：专辑10：预留（内容集）11：图文单篇12：图文集13：直播节目
	 * @author douzi
	 * @param vo
	 */
    public void process(OutSourceDataParamsVo paramVo) {
        try {
			// 获取bcId
        	Long bcId = getBcId(paramVo);
        	
        	// 发送kafka
        	sendKafkaProcess(paramVo, bcId);
        } catch(Exception e) {
        	log.info("播控发送kafka业务错误： ", e);
        } finally {
        	
        }
    }

    /**
     * <p><b>Title:</b> getBcId</p>
     * <p><b>Description:</b> 获取播控方id；出自poms老代码</p>
     * @author douzi
     * @param paramVo
     * @return
     */
	private Long getBcId(OutSourceDataParamsVo paramVo) {
		String forcePrdInfoBcIds = outHttpUtil.findByKey("IS_FORCE_PACKAGE_BC");
		
		PrdInfo prdInfo = prdInfoService.findById(paramVo.getPrdInfoId());
		Content content = contentService.findByContentId(paramVo.getContentId() + "");
		String superBcType = "";
		Long bcId = null;
		if (paramVo.getAssetType() == 7) {
			Content	parentContent = contentService.findByContentId(content.getSerialContentId());
			if (parentContent != null) {
				superBcType = parentContent.getSuperBcType();
			}
			if (parentContent != null && StrUtil.isNotBlank(parentContent.getSuperBcId())) {
				if (!forcePrdInfoBcIds.contains(StrUtil.nullToEmpty(prdInfo.getBcId() + ""))) {// 不匹配强制商品包播控
					bcId = Long.parseLong(parentContent.getSuperBcId());
				} else {
					bcId = prdInfo.getBcId();
				}
			} else {
				bcId = prdInfo.getBcId();
			}
		} else {
			superBcType = content.getSuperBcType();
			if (StrUtil.isNotBlank(content.getSuperBcId())) {
				if (!forcePrdInfoBcIds.contains(StrUtil.nullToEmpty(prdInfo.getBcId() + ""))) {// 不匹配强制商品包播控
					bcId = Long.parseLong(content.getSuperBcId());
				} else {
					bcId = prdInfo.getBcId();
				}
			} else {// 内容上无播控方
				bcId = prdInfo.getBcId();
			}
		}
		
		return bcId;
	}
	
	/**
     * <p><b>Title:</b> sendKafka</p>
     * <p><b>Description:</b> 发送kafka的流程、判断是否需要播控</p>
     *
     * @param e
     * @param br
     * @author douzi
     */
    void sendKafkaProcess(OutSourceDataParamsVo paramVo, Long bcId) {
        log.info("主流程|判断是否直播节目单PlayBillFlag:" + paramVo.getPlayBillFlag() + "|媒资状态：" + paramVo.getStatus() + "|播控内容：" + paramVo.getContentId());
        if (Integer.valueOf(1).equals(paramVo.getPlayBillFlag())) {
            sendKafkaPre(paramVo, bcId);
            return;
        }

        switch (paramVo.getStatus()) {
            case Constants.ASSET_STATUS_0:
                sendKafkaPre(paramVo, bcId);
                break;
            case Constants.ASSET_STATUS_1:
//			case Constants.ASSET_STATUS_2:
            case Constants.ASSET_STATUS_3:
                if (isNeedBc(paramVo)) {
                    sendKafkaPre(paramVo, bcId);
                }
                break;
            default:
                log.info("媒资:" + paramVo.getAssetId() + "|内容:" + paramVo.getContentId() + "|不属于任何一种媒资状态:" + paramVo.getStatus());
                break;
        }
    }

    public void sendKafkaPre(OutSourceDataParamsVo paramsVo, Long bcId) {
        if (!"on".equalsIgnoreCase(outHttpUtil.findByKey("KAFKA_SWITCH"))) { //kafka 开关
            return;
        }
        this.sendKafka(paramsVo, bcId);
    }

    /**
     * <p><b>Title:</b> sendKafka</p>
     * <p><b>Description:</b> 发送kafka主体</p>
     *
     * @param paramsVo
     * @author douzi
     */
    public void sendKafka(OutSourceDataParamsVo paramsVo, Long bcId) {
        String kafkaToptic = getKafkaToptic(paramsVo, bcId);
        if (StringUtils.isBlank(kafkaToptic)) {
            log.info("节目id:|{}|媒资id:|{}|没有配置kafka topic", paramsVo.getPrdContId(), paramsVo.getAssetId());
            return;
        }

        List<ProductContentPojo> productAssetPojos = prdContService.getbcListByPrdContId(paramsVo.getPrdContId(), bcId);
        if (productAssetPojos == null) {
            return;
        }
        String bcXml = getBcXml(productAssetPojos);
        log.info("PrdContId:" + paramsVo.getPrdContId() + "|productSecondId:" + paramsVo.getProductSecondId() + "|bcXml:" + bcXml);
        if (StringUtils.isBlank(bcXml)) {
            return;
        }
//        kafkaProducer.send(kafkaToptic, br.getPrdContId().toString(), bcXml);
//		messageSender.send(bcXml, kafkaToptic);

        log.info("发送kafka的流程、判断是否需要播控|topic|{}|message|{}", kafkaToptic, bcXml);
//        kafkaFeignClient.kafkaSend(kafkaToptic, bcXml);
        //自动发送节目单给播控方
//        sendPlayBillKafka(paramsVo);
    }

    /**
     * <p><b>Title:</b> sendPlayBillKafka</p>
     * <p><b>Description:</b>自动发送节目单</p>
     *
     * @param paramsVo
     * @author huzhenjie
     */
//    public void sendPlayBillKafka(OutSourceDataParamsVo paramsVo) {
//        Integer assetType = paramsVo.getAssetType();
//        String type = null;
//        if (assetType != null) {
//            type = assetType.toString();
//        }
//        log.info("是不是直播类型type|" + type);
//        if ("15".equals(type)) {
//            log.info("是直播类型,装备发送kafka|" + paramsVo.getAssetId().toString());
//            outHttpUtil.AutoSendPlayBill(paramsVo.getAssetId().toString());
//            return;
//        }
//
//        String fromType = null;
//        if (paramsVo.getType() != null) {
//            fromType = paramsVo.getType().toString();
//        }
//        log.info("是不是直播类型fromType|" + fromType);
//
//        if (("8".equals(fromType) || "9".equals(fromType) || "12".equals(fromType))) {
//            log.info("是直播类型,装备发送kafka|" + paramsVo.getAssetId().toString());
//            outHttpUtil.AutoSendPlayBill(paramsVo.getAssetId().toString());
//        }
//
//
//    }


    /**
     * <p><b>Title:</b> getKafkaToptic</p>
     * <p><b>Description:</b> 获取kafka toptic</p>
     *
     * @param paramsVo
     * @return
     * @author douzi
     */
    private String getKafkaToptic(OutSourceDataParamsVo paramsVo, Long bcId) {
        String kafkaToptic = "";
        if (null != bcId) {
            //bcId对应key,value对应topic
            kafkaToptic = outHttpUtil.findByKey(bcId.toString());
        }
        log.info("PrdContId:" + paramsVo.getPrdContId() + "|productSecondId:" + paramsVo.getProductSecondId() + "|bcId:" + bcId + "|kafkaToptic:" + kafkaToptic);
        return kafkaToptic;
    }

    /**
     * <p><b>Title:</b> getBcXml</p>
     * <p><b>Description:</b> 获取模板xml</p>
     *
     * @param productAssetPojos
     * @return
     * @author douzi
     */
    private String getBcXml(List<ProductContentPojo> productAssetPojos) {
        Map<String, Object> tmpMap = new HashMap<String, Object>();
        tmpMap.put("bclist", productAssetPojos);
        tmpMap.put("resCode", "000000");
        tmpMap.put("resMessage", "");

        Map<String, Object> metaMap = new HashMap<String, Object>();
        metaMap.put("poms", tmpMap);
        String bcXml = "";
        ByteArrayOutputStream out = null;
        try {
            out = ExportXmlUtil.doExport(metaMap, FtlConstants.POMS_INTERFACE);
            bcXml = out.toString();
        } catch (Exception e) {
            throw new BusinessException(MessageCodes.M333333);
        } finally {
            IoUtil.close(out);
        }
        return bcXml;
    }

    /**
     * <p><b>Title:</b> isNeedBc</p>
     * <p><b>Description:</b> 判断源数据是否需要再次播控</p>
     *
     * @param paramVo
     * @author douzi
     */
    boolean isNeedBc(OutSourceDataParamsVo paramVo) {
        if (StringUtils.isNotBlank(paramVo.getUpdateFields())) {
            //NEED_BC_FIELDS="1,3,5,7"
            String needBcFields = outHttpUtil.findByKey("NEED_BC_FIELDS");
            log.info("PrdContId:" + paramVo.getPrdContId() + "|productSecondId:" + paramVo.getProductSecondId() + "|needBcFields:" + needBcFields + "|updateFields:" + paramVo.getUpdateFields());
            String[] bcFields = needBcFields.split(",");
            String[] updateFields = paramVo.getUpdateFields().split(",");
            //判断变更信息中是否包含需要播控的字段
            for (String need : bcFields) {
                for (String updateField : updateFields) {
                    if (updateField.equals(need)) {
                        return true;
                    }
                }
            }
        }
        return false;
    }
}
